package spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

object Transform {
  // Todo DStream转换
  /**
   * DStream 上的操作与RDD的类似,分为Transformations（转换）和Output Operations（输出）两种，
   * 此外转换操作中还有一些比较特殊的原语，如：updateStateByKey()、transform() 以及各种Window相关的原语。
   * */
  // Todo 1. 无状态转换
  /**
   * 无状态转化操作就是把简单的RDD转化操作应用到每个批次上，也就是转化DStream中的每一个RDD。
   * 但注意：其返回类型不是RDD而依然是DStream
   * 尽管这些函数看起来像作用在整个流上一样，但事实上每个DStream在内部是由许多RDD（批次）组成，且无状态转化操作是分别应用到每个RDD上的。
   * 例如：reduceByKey()会归约每个时间区间中的数据，但不会归约不同区间之间的数据。
   * */
  // Todo 2. 有状态转换
  /**
   * 1. UpdateStateByKey 原语
   *      用于记录历史记录，有时，我们需要在DStream中跨批次维护状态(例如流计算中累加wordCount)。
   *      它的结果会是一个新的DStream，其内部的RDD 序列是由每个时间区间对应的(键，状态)对组成的。
   *      参数：
   *        1. 定义状态，状态可以是一个任意的数据类型。
   *        2. 定义状态更新函数，用此函数阐明如何使用之前的状态和来自输入流的新值对状态进行更新。
   *          使用updateStateByKey 需要对检查点目录进行配置，会使用检查点来保存状态。
   * */
  def main(args: Array[String]): Unit = {
    //创建SparkConf
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")

    //创建StreamingContext
    val ssc: StreamingContext = new StreamingContext(sparkConf, Seconds(3))

    //创建DStream
    val lineDStream: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 9999)

    //转换为RDD操作 该函数每一批次调度一次！
    val wordAndCountDStream: DStream[(String, Int)] = lineDStream.transform(
      rdd => {
        val words: RDD[String] = rdd.flatMap(_.split(" "))

        val wordAndOne: RDD[(String, Int)] = words.map((_, 1))

        val value: RDD[(String, Int)] = wordAndOne.reduceByKey(_ + _)

        value
      }
    )

    // print
    wordAndCountDStream.print()

    // start and wait for termination
    ssc.start()
    ssc.awaitTermination()

  }
}
